"""
@File    : utils.py
@Time    : 2025/7/17 上午9:04
@Author  : xiashuobad
@Desc    : 

/**
 * _ooOoo_
 * o8888888o
 * 88" . "88
 * (| -_- |)
 *  O\ = /O
 * ___/`---'\____
 * .   ' \\| |// `.
 * / \\||| : |||// \
 * / _||||| -:- |||||- \
 * | | \\\ - /// | |
 * | \_| ''\---/'' | |
 * \ .-\__ `-` ___/-. /
 * ___`. .' /--.--\ `. . __
 * ."" '< `.___\_<|>_/___.' >'"".
 * | | : `- \`.;`\ _ /`;.`/ - ` : | |
 * \ \ `-. \_ __\ /__ _/ .-` / /
 * ======`-.____`-.___\_____/___.-`____.-'======
 * `=---=' bug泛滥 佛已瘫痪
"""
# rsio/core/utils.py
import contextlib
import logging
import itertools
from functools import wraps
from importlib.util import find_spec
from importlib import import_module
from rich.logging import RichHandler
from collections import abc


class RichLogger:
    def __init__(self, logger: logging.Logger):
        self.logger = logger

    def __getattr__(self, attr):
        return getattr(self.logger, attr)

    def debug(self, msg: str, markup: str = None):
        if markup:
            self.logger.debug(f'[{markup}]{msg}', extra={'markup': True})
        else:
            self.logger.debug(msg)

    def info(self, msg: str, markup: str = None):
        if markup:
            self.logger.info(f'[{markup}]{msg}', extra={'markup': True})
        else:
            self.logger.info(msg)

    def warning(self, msg: str, markup: str = None):
        if markup:
            self.logger.warning(f'[{markup}]{msg}', extra={'markup': True})
        else:
            self.logger.warning(msg)

    def error(self, msg: str, markup: str = None):
        if markup:
            self.logger.error(f'[{markup}]{msg}', extra={'markup': True})
        else:
            self.logger.error(msg)

    def critical(self, msg: str, markup: str = None):
        if markup:
            self.logger.critical(f'[{markup}]{msg}', extra={'markup': True})
        else:
            self.logger.critical(msg)

    def exception(self, msg: str, markup: str = None):
        if markup:
            self.logger.exception(f'[{markup}]{msg}', extra={'markup': True})
        else:
            self.logger.exception(msg)


def setup_logger(name='rsio', level=logging.INFO, log_file=None):
    logger = logging.getLogger(name)
    logger.setLevel(level)
    log_formatter = logging.Formatter('%(asctime)s [%(levelname)s] %(name)s: %(message)s')

    if not logger.handlers and not logger.parent.handlers:
        # 控制台输出
        # ch = logging.StreamHandler()
        ch = RichHandler(rich_tracebacks=True)
        ch.setLevel(level)
        ch.setFormatter(log_formatter)
        logger.addHandler(ch)
    # 文件输出（可选）
    if log_file:
        for handler in logger.handlers:
            if isinstance(handler, logging.FileHandler):
                handler.close()
                logger.removeHandler(handler)
        fh = logging.FileHandler(log_file)
        fh.setLevel(level)
        fh.setFormatter(log_formatter)
        logger.addHandler(fh)

    return RichLogger(logger)


logger = setup_logger(__name__)


@contextlib.contextmanager
def catch_exception(raise_exception: Exception = None):
    """
    上下文管理器，捕获异常，按自定义的异常抛出
    :param raise_exception: 自定义异常的实例
    :return:
    """
    try:
        yield
    except Exception as e:
        logger.exception(str(e), markup='red')
        if raise_exception:
            new_args = f"{''.join(raise_exception.args)}，具体报错信息为：{str(e)}"
            raise_exception = type(raise_exception)(new_args)
            raise raise_exception
        raise e


def try_import(*module_parts: str):
    """
    尝试导入一个模块或子模块，支持嵌套导入。

    示例:
        gdal = try_import('osgeo', 'gdal')  # 等价于 from osgeo import gdal
        np = try_import('numpy')            # 等价于 import numpy

    返回:
        模块对象 或 None（如果未找到）
    """
    parent_path = module_parts[:-1]
    module_name = module_parts[-1]

    if not parent_path:
        # 单层导入：import xxx
        spec = find_spec(module_name)
        if spec is None:
            return None
        return import_module(module_name)

    # 多层导入：from xxx import yyy
    parent_module = import_module('.'.join(parent_path))
    if not hasattr(parent_module, module_name):
        return None

    return getattr(parent_module, module_name)


def iter_cast(inputs, dst_type, return_type=None):
    """
    该函数用于将可迭代对象中的元素转换为指定的类型。函数接受三个参数
    :param inputs: 可迭代对象
    :param dst_type: 目标类型
    :param return_type: 可选参数,指定输出对象的类型
    :return: 如果指定了return_type，则返回转换后的指定类型对象，否则返回转换后的迭代器。
    """
    if not isinstance(inputs, abc.Iterable):
        raise TypeError('inputs参数必须是一个可迭代对象!')
    if not isinstance(dst_type, type):
        raise TypeError('dst_type参数必须是一个有效的类型!')

    out_iterable = map(dst_type, inputs)

    if return_type is None:
        return out_iterable
    else:
        return return_type(out_iterable)


def list_cast(inputs, dst_type):
    """
    该函数用于将可迭代对象中的元素转换为指定的类型，并返回一个列表。
    :param inputs: 可迭代对象
    :param dst_type: 目标类型
    :return: 返回一个list列表
    """
    return iter_cast(inputs, dst_type, return_type=list)


def tuple_cast(inputs, dst_type):
    """
    该函数用于将可迭代对象中的元素转换为指定的类型，并返回一个元组。
    :param inputs: 可迭代对象
    :param dst_type: 目标类型
    :return: 返回一个元组
    """
    return iter_cast(inputs, dst_type, return_type=tuple)


def is_seq_of(seq, expected_type, seq_type=None):
    """
    检查序列是否由特定类型元素组成。
    :param seq: 待检查的序列。
    :param expected_type: 期望的元素类型。
    :param seq_type: (可选) 序列的类型。如果未提供，则默认为任何序列。
    :return: 如果序列是由期望的类型组成的，则返回True；否则返回False。
    """

    if seq_type is None:
        exp_seq_type = abc.Sequence
    else:
        assert isinstance(seq_type, type)
        exp_seq_type = seq_type
    if not isinstance(seq, exp_seq_type):  # 如果序列不是期望的类型，则返回False
        return False
    for item in seq:  # 遍历序列中的每个元素
        if not isinstance(item, expected_type):  # 如果元素不是期望的类型，则返回False
            return False
    return True


def is_list_of(seq, expected_type: (type, list, tuple)):
    """
    检查序列是否是列表类型，并且里面的元素是否是预期的类型。
    :param seq: 待检查的序列
    :param expected_type: 期望的元素类型。阔以是一个列表或元组，只要是其中的一个类型满足即可
    :return: 如果序列是列表类型，并且里面的元素是预期的类型，则返回True；否则返回False。
    """
    if isinstance(expected_type, type):
        expected_type = [expected_type]
    return any([is_seq_of(seq, expected_type_i, seq_type=list) for expected_type_i in expected_type])


def is_tuple_of(seq, expected_type: (type, list, tuple)):
    """
    检查序列是否是元组类型，并且里面的元素是否是预期的类型。
    :param seq: 待检查的序列
    :param expected_type: 期望的元素类型。
    :return: 如果序列是元组类型，并且里面的元素是预期的类型，则返回True；否则返回False。
    """
    if isinstance(expected_type, type):
        expected_type = [expected_type]
    return any([is_seq_of(seq, expected_type_i, seq_type=tuple) for expected_type_i in expected_type])


def slice_list(in_list, lens):
    """
    该函数用于将一个列表按照指定的分割长度切分成多个子列表
    如果lens是单个整数，列表将被均分为长度为lens的多个子列表。
    如果lens是整数列表，列表将根据这些整数作为长度进行分割。
    :param in_list: 待分割的列表。
    :param lens: 分割长度，可以是单个整数或整数列表。
    :return: 一个由按指定长度分割的子列表组成的列表。
    """
    if isinstance(lens, int):
        assert len(in_list) % lens == 0
        lens = [lens] * int(len(in_list) / lens)
    if not isinstance(lens, list):
        raise TypeError('"indices" must be an integer or a list of integers')
    elif sum(lens) != len(in_list):
        raise ValueError('sum of lens and list length does not '
                         f'match: {sum(lens)} != {len(in_list)}')
    out_list = []
    idx = 0
    for i in range(len(lens)):
        out_list.append(in_list[idx:idx + lens[i]])
        idx += lens[i]
    return out_list


def concat_list(in_list):
    """
    将嵌套列表中的所有元素连接成一个平铺的列表。
    :param in_list:一个嵌套的列表，其中包含多个子列表。
    :return:一个平铺的列表，包含嵌套列表中的所有元素。
    """
    # 使用itertools.chain函数将嵌套列表中的所有子列表连接成一个平铺的列表
    # itertools.chain函数的作用是将多个迭代器对象组合成一个单一的迭代器
    return list(itertools.chain(*in_list))


def cost_time(name: str = ''):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            import time
            start = time.time()
            result = func(*args, **kwargs)
            end = time.time()
            logger.info(f"{name or func.__name__} 耗时：{end - start:.2f}s")
            return result

        return wrapper

    return decorator
